package org.apache.spark.examples

import org.apache.spark.{SparkConf, SparkContext}

object RDDMapOperations {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("RDD Map Operations").setMaster("local[1]")
    val sc = new SparkContext(conf)

    // 1. 创建一个简单的数字RDD
    val numbersRDD = sc.parallelize(1 to 10, 2)
    println("\n=== 原始数据 ===")
    numbersRDD.collect().foreach(x => print(s"$x "))
    println()

    // 2. 基础map操作：将每个数字翻倍
    val doubledRDD = numbersRDD.map(_ * 2)
    println("\n=== 每个数字翻倍 ===")
    doubledRDD.collect().foreach(x => print(s"$x "))
    println()

    // 3. map with 条件判断：偶数翻倍，奇数加1
    val conditionalRDD = numbersRDD.map(x => if (x % 2 == 0) x * 2 else x + 1)
    println("\n=== 条件处理：偶数翻倍，奇数加1 ===")
    conditionalRDD.collect().foreach(x => print(s"$x "))
    println()

    // 4. map with 字符串转换：将数字转换为字符串格式
    val stringRDD = numbersRDD.map(x => s"Number: $x")
    println("\n=== 转换为字符串 ===")
    stringRDD.collect().foreach(println)

    // 5. map with 自定义对象：将数字转换为简单的对象
    case class NumberInfo(value: Int, isEven: Boolean, square: Int)
    val objectRDD = numbersRDD.map(x => NumberInfo(x, x % 2 == 0, x * x))
    println("\n=== 转换为对象 ===")
    objectRDD.collect().foreach(x => println(s"值: ${x.value}, 是否偶数: ${x.isEven}, 平方: ${x.square}"))

    // 6. map with 数学运算：计算每个数字的平方根
    val sqrtRDD = numbersRDD.map(x => math.sqrt(x))
    println("\n=== 计算平方根 ===")
    sqrtRDD.collect().foreach(x => println(f"$x%.2f"))

    // 7. map with 多值转换：将每个数字转换为(原值，平方，立方)的元组
    val multiValueRDD = numbersRDD.map(x => (x, x * x, x * x * x))
    println("\n=== 多值转换 ===")
    multiValueRDD.collect().foreach { case (original, square, cube) =>
      println(f"原值: $original, 平方: $square, 立方: $cube")
    }

    // 8. map with 格式化：将数字格式化为固定长度的字符串
    val formattedRDD = numbersRDD.map(x => f"$x%03d")
    println("\n=== 数字格式化 ===")
    formattedRDD.collect().foreach(println)

    // 9. map with 范围检查：将数字映射到指定范围(1-5)内
    val rangeRDD = numbersRDD.map(x => math.min(5, math.max(1, x)))
    println("\n=== 范围限制(1-5) ===")
    rangeRDD.collect().foreach(x => print(s"$x "))
    println()

    // 10. map with 异常处理：处理可能的异常情况
    val safeRDD = numbersRDD.map(x => {
      try {
        if (x > 5) throw new IllegalArgumentException("Too big!")
        x
      } catch {
        case _: IllegalArgumentException => 5 // 超过5的数字都设为5
      }
    })
    println("\n=== 异常处理 ===")
    safeRDD.collect().foreach(x => print(s"$x "))
    println()

    // 暂停以便查看Spark UI
    Thread.sleep(300000)

    sc.stop()
  }
} 